package com.raft.core.node.task;

import java.util.concurrent.Callable;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.raft.core.node.NodeEndpoint;
import com.raft.core.node.NodeId;
import com.raft.core.node.config.NodeConfig;
import com.raft.core.rpc.message.AppendEntriesResultMessage;
import com.raft.core.rpc.message.InstallSnapshotResultMessage;
import com.raft.core.rpc.message.InstallSnapshotRpc;

public class NewNodeCatchUpTask implements Callable<NewNodeCatchUpTaskResult> {

	// 任务状态
    private enum State {
        START,
        REPLICATING,
        REPLICATION_FAILED,
        REPLICATION_CATCH_UP,
        TIMEOUT
    }

    private static final Logger logger = LoggerFactory.getLogger(NewNodeCatchUpTask.class);
    private final NewNodeCatchUpTaskContext context;
    private final NodeEndpoint endpoint;
    private final NodeId nodeId;
    private final NodeConfig config;
    private State state = State.START;
    private boolean done = false;
    private long lastReplicateAt; // set when start
    private long lastAdvanceAt; // set when start
    private int round = 1;
    private int nextIndex = 0; // reset when receive append entries result
    private int matchIndex = 0;

    public NewNodeCatchUpTask(NewNodeCatchUpTaskContext context, NodeEndpoint endpoint, NodeConfig config) {
        this.context = context;
        this.endpoint = endpoint;
        this.nodeId = endpoint.getId();
        this.config = config;
    }

    public NodeId getNodeId() {
        return nodeId;
    }

    // 主函数，在操作线程中执行
    @Override
    public synchronized NewNodeCatchUpTaskResult call() throws Exception {
        logger.info("task start");
        setState(State.START);
        // 同步日志
        context.replicateLog(endpoint);
        lastReplicateAt = System.currentTimeMillis();
        lastAdvanceAt = lastReplicateAt;
        setState(State.REPLICATING);
        // 是否结束
        // 结束有多种可能，比如catch-up完毕；无法继续复制；超时次数等
        while (!done) {
        	// 模拟超时
            wait(config.getNewNodeReadTimeout());
            // 有可能成功；也有可能超时
            // 1. done
            // 2. replicate -> no response within timeout
            if (System.currentTimeMillis() - lastReplicateAt >= config.getNewNodeReadTimeout()) {
                logger.info("node {} not response within read timeout", endpoint.getId());
                state = State.TIMEOUT;
                break;
            }
        }
        // 理论上，catch-up是leader节点和新节点一对一同步的过程，可以设计成一个单线程处理的方式。但是那样的话，通信部分需要改成阻塞式的，无法复用现有的基于nio的通信组件。
        // 如果愿意再写一个阻塞式的通信部分的话，可以不需要TaskGroup的消息转发，直接在Task内部处理即可。
        // call方法中的done主要有以下可能性：
        // 1、catch-up完成
        // 2、超过最大不同来回，比如10次
        // 3、日志匹配失败，无法继续同步
        // 4、指定时间内同步太慢(慢网速等情况)
        // wait方法用来模拟节点的超时，模拟是因为nio没有超时的概念。如果距离发送时间太长，就直接认为超时。具体来说，从context.replicateLog发送日志同步消息开始，wait方法
        // 释放实例锁。如果指定时间内没有收到响应，wait重新获取实例锁，并进入超时流程。
        // 进入wait方法后，任务有机会获取实例锁处理来自新节点的影响，典型的是AppendEntries消息的处理
        logger.info("task done");
        // 任务完成
        context.done(this);
        return mapResult(state);
    }

    /**映射结果
     * @param state
     * @return
     */
    private NewNodeCatchUpTaskResult mapResult(State state) {
        switch (state) {
            case REPLICATION_CATCH_UP:
                return new NewNodeCatchUpTaskResult(nextIndex, matchIndex);
            case REPLICATION_FAILED:
                return new NewNodeCatchUpTaskResult(NewNodeCatchUpTaskResult.State.REPLICATION_FAILED);
            default:
                return new NewNodeCatchUpTaskResult(NewNodeCatchUpTaskResult.State.TIMEOUT);
        }
    }

    private void setState(State state) {
        logger.info("state -> {}", state);
        this.state = state;
    }

    // 此方法在主线程中执行
    // in node thread
    synchronized void onReceiveAppendEntriesResult(AppendEntriesResultMessage resultMessage, int nextLogIndex) {
        assert nodeId.equals(resultMessage.getSourceNodeId());
        if (state != State.REPLICATING) {
            throw new IllegalStateException("receive append entries result when state is not replicating");
        }
        // 第一次收到消息时，设置nextIndex
        if (nextIndex == 0) {
            nextIndex = nextLogIndex;
        }
        logger.info("replication state of new node {}, next index {}, match index {}", nodeId, nextIndex, matchIndex);
        if (resultMessage.get().isSuccess()) {
            int lastEntryIndex = resultMessage.getRpc().getLastEntryIndex();
            assert lastEntryIndex >= 0;
            matchIndex = lastEntryIndex;
            nextIndex = lastEntryIndex + 1;
            lastAdvanceAt = System.currentTimeMillis();
            // catch up完成
            if (nextIndex >= nextLogIndex) {
                setStateAndNotify(State.REPLICATION_CATCH_UP);
                return;
            }
            // 超时交互次数
            if ((++round) > config.getNewNodeMaxRound()) {
                logger.info("node {} cannot catch up within max round", nodeId);
                setStateAndNotify(State.TIMEOUT);
                return;
            }
        } else {
        	// 无法继续同步
            if (nextIndex <= 1) {
                logger.warn("node {} cannot back off next index more, stop replication", nodeId);
                setStateAndNotify(State.REPLICATION_FAILED);
                return;
            }
            nextIndex--;
            // 慢速网络等
            if (System.currentTimeMillis() - lastAdvanceAt >= config.getNewNodeAdvanceTimeout()) {
                logger.info("node {} cannot make progress within timeout", nodeId);
                setStateAndNotify(State.TIMEOUT);
                return;
            }
        }
        // 接着同步日志
        context.doReplicateLog(endpoint, nextIndex);
        lastReplicateAt = System.currentTimeMillis();
        // 通知wait，继续下一个同步的超时检查
        notify();
    }

    // 对于安装日志快照消息来说，全部消息传输完毕才算安装完成，并且需要区分使用普通日志复制和日志快照安装的消息
    // in node thread
    synchronized void onReceiveInstallSnapshotResult(InstallSnapshotResultMessage resultMessage, int nextLogIndex) {
        assert nodeId.equals(resultMessage.getSourceNodeId());
        if (state != State.REPLICATING) {
            throw new IllegalStateException("receive append entries result when state is not replicating");
        }
        InstallSnapshotRpc rpc = resultMessage.getRpc();
        // 日志安装完成
        if (rpc.isDone()) {
            matchIndex = rpc.getLastIndex();
            nextIndex = rpc.getLastIndex() + 1;
            lastAdvanceAt = System.currentTimeMillis();
            if (nextIndex >= nextLogIndex) {
                setStateAndNotify(State.REPLICATION_CATCH_UP);
                return;
            }
            // 增加回数
            round++;
            // 继续复制
            context.doReplicateLog(endpoint, nextIndex);
        } else {
        	// 日志安装
            context.sendInstallSnapshot(endpoint, rpc.getOffset() + rpc.getDataLength());
        }
        lastReplicateAt = System.currentTimeMillis();
        notify();
    }

    private void setStateAndNotify(State state) {
        setState(state);
        done = true;
        notify();
    }

    @Override
    public String toString() {
        return "NewNodeCatchUpTask{" +
                "state=" + state +
                ", endpoint=" + endpoint +
                ", done=" + done +
                ", lastReplicateAt=" + lastReplicateAt +
                ", lastAdvanceAt=" + lastAdvanceAt +
                ", nextIndex=" + nextIndex +
                ", matchIndex=" + matchIndex +
                ", round=" + round +
                '}';
    }

}
